package mapreduce;

import static util.Preconditions.checkNotNull;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

import util.CharacterSet;

public class AssociationRulesReducer extends MapReduceBase implements Reducer<CharacterSet, LongWritable, CharacterSet, LongWritable> {

	private long minOccurrences;

	@Override
	public void configure(JobConf configuration) {
		minOccurrences = Long.parseLong(configuration.get("MIN_OCCURRENCES"));
	}

	@Override
	public void close() throws IOException {
		// TODO Auto-generated method stub

	}

	@Override
	public void reduce(CharacterSet key, Iterator<LongWritable> values,
			OutputCollector<CharacterSet, LongWritable> output, Reporter reporter)
					throws IOException {
		checkNotNull(key, "key must not be null");
		checkNotNull(values, "values must not be null");
		checkNotNull(output, "outputCollector must not be null");
		checkNotNull(reporter, "reporter must not be null");

		Long newKey = 0L;

		while (values.hasNext()) {
			newKey += values.next().get();
		}

		if (newKey >= minOccurrences) {
			output.collect(key, new LongWritable(newKey));
		}
	}
}
